package org.apache.flink.runtime.allocatedeploy.execution;

import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;

import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;

import javax.annotation.Nullable;

import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.ErrorInfo;
import org.apache.flink.runtime.ResourceException;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.runtime.consumermanager.slots.ProviderManagerGateway;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.providermanager.ProviderManagerLocation;
import org.apache.flink.runtime.resourceconsumer.ConsumingMasterId;
import org.apache.flink.runtime.resourceconsumer.LogicalSlot;
import org.apache.flink.util.ERException;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.concurrent.FutureUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @类名: BusinessExecution
 * @说明: 执行
 *       封装执行命令，执行参数，所需资源
 *
 * @author   leehom
 * @Date	 2024年7月25日 下午5:22:02
 * 修改记录：
 *
 * @see 	 
 */
public class Execution implements LogicalSlot.Payload {

    static final Logger LOG = LoggerFactory.getLogger(Execution.class);

    private static final int NUM_CANCEL_CALL_TRIES = 3;

    // --------------------------------------------------------------------------------------------

    /** The executor which is used to execute futures. */
    private final Executor executor;
	/** 用户自定义*/
	private String commandKey;
	/** 用户自定义*/
	private String command;
	/** 序列化的参数*/
	private String serArgs;
	/** 资源key，用户定义，获取对应的ResourceProfile*/
	private String resourceKey;
    /** The ID referencing the attempt to execute the task. */
    private ExecutionID executionId;
    /** id of the master of consuming */
    private ConsumingMasterId consumingMasterId;
    /** */
    private ConsumerID consumerId;

    /**
     * The timestamps when state transitions occurred, indexed by {@link ExecutionState#ordinal()}.
     */
    private final long[] stateTimestamps;

    /**
     * The end timestamps when state transitions occurred, indexed by {@link
     * ExecutionState#ordinal()}.
     */
    private final long[] stateEndTimestamps;

    private final Time rpcTimeout;

    /** A future that completes once the BusinessExecution reaches a terminal ExecutionState. */
    private final CompletableFuture<ExecutionState> terminalStateFuture;

    private final CompletableFuture<?> releaseFuture;

    private final CompletableFuture<ProviderManagerLocation> providerManagerLocationFuture;
    
    private final ComponentMainThreadExecutor mainThreadExecutor;

    /**
     * Gets completed successfully when the task switched to {@link ExecutionState#INITIALIZING} or
     * {@link ExecutionState#RUNNING}. If the task never switches to those state, but fails
     * immediately, then this future never completes.
     */
    private final CompletableFuture<?> initializingOrRunningFuture;

    private volatile ExecutionState state = ExecutionState.CREATED;

    private LogicalSlot assignedResource;

    private Optional<ErrorInfo> failureCause = Optional.empty(); // once an ErrorInfo is set, never changes

    /** This field holds the allocation id once it was assigned successfully. */
    @Nullable private AllocationID assignedAllocationID;

    /**
     * Creates a new BusinessExecution.
     *
     * @param executor The executor used to dispatch callbacks from futures and asynchronous RPC
     *     calls.
     * @param vertex The execution vertex to which this BusinessExecution belongs
     * @param attemptNumber The execution attempt number.
     * @param startTimestamp The timestamp that marks the creation of this BusinessExecution
     * @param rpcTimeout The rpcTimeout for RPC calls like deploy/cancel/stop.
     */
    public Execution(
    		ConsumerID consumerId,
    		String commandKey,
    		String command,
    		String serArgs,
    		String resourceKey,
            Executor executor,
            long startTimestamp,
            Time rpcTimeout,
            ComponentMainThreadExecutor mainThreadExecutor) {
    	this(
    			commandKey,
        		command,
        		serArgs,
                executor,
                startTimestamp,
                rpcTimeout,
                mainThreadExecutor);
    	this.resourceKey = resourceKey;
    	this.consumerId = consumerId;
    }
    
    public Execution(
    		String commandKey,
    		String command,
    		String serArgs,
            Executor executor,
            long startTimestamp,
            Time rpcTimeout,
            ComponentMainThreadExecutor mainThreadExecutor) {
    	// 执行命令，参数
    	this.commandKey = commandKey;
    	this.command = command;
    	this.serArgs = serArgs;
    	
        this.executor = checkNotNull(executor);
        this.executionId = new ExecutionID();
        this.rpcTimeout = checkNotNull(rpcTimeout);
        this.mainThreadExecutor = mainThreadExecutor;

        this.stateTimestamps = new long[ExecutionState.values().length];
        this.stateEndTimestamps = new long[ExecutionState.values().length];
        markTimestamp(ExecutionState.CREATED, startTimestamp);

        this.terminalStateFuture = new CompletableFuture<>();
        this.releaseFuture = new CompletableFuture<>();
        this.providerManagerLocationFuture = new CompletableFuture<>();
        this.initializingOrRunningFuture = new CompletableFuture<>();

        this.assignedResource = null;
    }

    // --------------------------------------------------------------------------------------------
    //   Properties
    // --------------------------------------------------------------------------------------------
    public ExecutionID getExecutionId() {
        return executionId;
    }

    public ExecutionState getState() {
        return state;
    }

    @Nullable
    public AllocationID getAssignedAllocationID() {
        return assignedAllocationID;
    }

    public CompletableFuture<ProviderManagerLocation> getProviderManagerLocationFuture() {
        return providerManagerLocationFuture;
    }

    public LogicalSlot getAssignedResource() {
        return assignedResource;
    }

    /**
     * 
     * 尝试分配资源slot给Execution。
     *
     * @param logicalSlot to assign to this execution
     * @return true if the slot could be assigned to the execution, otherwise false
     */
    public boolean tryAssignResource(final LogicalSlot logicalSlot) {

        assertRunningInResourceConsumerMainThread();

        checkNotNull(logicalSlot);

        // only allow to set the assigned resource in state SCHEDULED or CREATED
        // note: we also accept resource assignment when being in state CREATED for testing purposes
        if (state == ExecutionState.SCHEDULED || state == ExecutionState.CREATED) {
            if (assignedResource == null) {
                assignedResource = logicalSlot;
                if (logicalSlot.tryAssignPayload(this)) {
                    // check for concurrent modification (e.g. cancelling call)
                    if ((state == ExecutionState.SCHEDULED || state == ExecutionState.CREATED)
                            && !providerManagerLocationFuture.isDone()) {
                        providerManagerLocationFuture.complete(logicalSlot.getTaskManagerLocation());
                        assignedAllocationID = logicalSlot.getAllocationId();
                        return true;
                    } else {
                        // free assigned resource and return false
                        assignedResource = null;
                        return false;
                    }
                } else {
                    assignedResource = null;
                    return false;
                }
            } else {
                // the slot already has another slot assigned
                return false;
            }
        } else {
            // do not allow resource assignment if we are not in state SCHEDULED
            return false;
        }
    }

    public ProviderManagerLocation getAssignedResourceLocation() {
        // returns non-null only when a location is already assigned
        final LogicalSlot currentAssignedResource = assignedResource;
        return currentAssignedResource != null
                ? currentAssignedResource.getTaskManagerLocation()
                : null;
    }

    public Optional<ErrorInfo> getFailureInfo() {
        return failureCause;
    }

    public long[] getStateTimestamps() {
        return stateTimestamps;
    }

    public long[] getStateEndTimestamps() {
        return stateEndTimestamps;
    }

    public long getStateTimestamp(ExecutionState state) {
        return this.stateTimestamps[state.ordinal()];
    }

    public long getStateEndTimestamp(ExecutionState state) {
        return this.stateEndTimestamps[state.ordinal()];
    }

    public boolean isFinished() {
        return state.isTerminal();
    }

    /**
     * Gets a future that completes once the task execution reaches one of the states {@link
     * ExecutionState#INITIALIZING} or {@link ExecutionState#RUNNING}. If this task never reaches
     * these states (for example because the task is cancelled before it was properly deployed and
     * restored), then this future will never complete.
     *
     * <p>The future is completed already in the {@link ExecutionState#INITIALIZING} state, because
     * various running actions are already possible in that state (the task already accepts and
     * sends events and network data for task recovery). (Note that in earlier versions, the
     * INITIALIZING state was not separate but part of the RUNNING state).
     *
     * <p>This future is always completed from the job master's main thread.
     */
    public CompletableFuture<?> getInitializingOrRunningFuture() {
        return initializingOrRunningFuture;
    }

    /**
     * Gets a future that completes once the task execution reaches a terminal state. The future
     * will be completed with specific state that the execution reached. This future is always
     * completed from the job master's main thread.
     *
     * @return A future which is completed once the execution reaches a terminal state
     */
    @Override
    public CompletableFuture<ExecutionState> getTerminalStateFuture() {
        return terminalStateFuture;
    }

    /**
     * Gets the release future which is completed once the execution reaches a terminal state and
     * the assigned resource has been released. This future is always completed from the job
     * master's main thread.
     *
     * @return A future which is completed once the assigned resource has been released
     */
    public CompletableFuture<?> getReleaseFuture() {
        return releaseFuture;
    }

    /**
     * 部署执行到获分配的资源
     *
     * @throws JobException if the execution cannot be deployed to the assigned resource
     */
    public void deploy() throws ResourceException {
        assertRunningInResourceConsumerMainThread();

        // 局部变量
        final LogicalSlot slot = assignedResource;
        // 
        checkNotNull(
                slot,
                "In order to deploy the execution we first have to assign a resource via tryAssignResource.");

        // Check if the ProviderManager died in the meantime
        // This only speeds up the response to ProviderManager failing concurrently to deployments.
        // The more general check is the rpcTimeout of the deployment call
        if (!slot.isAlive()) {
            throw new ResourceException("Target slot (ProviderManager) for deployment is no longer alive.");
        }

        // make sure exactly one deployment call happens from the correct state
        ExecutionState previous = this.state;
        if (previous == ExecutionState.SCHEDULED) {
        	// 转换为部署中状态
            if (!transitionState(previous, ExecutionState.DEPLOYING)) {
                // race condition, someone else beat us to the deploying call.
                // this should actually not happen and indicates a race somewhere else
                throw new IllegalStateException(
                        "Cannot deploy task: Concurrent deployment call race.");
            }
        } else {
            // vertex may have been cancelled, or it was already scheduled
            throw new IllegalStateException(
                    "The BusinessExecution must be in SCHEDULED state to be deployed. Found state " + previous);
        }
        // 
        if (this != slot.getPayload()) {
            throw new IllegalStateException(
                    String.format(
                            "The execution %s has not been assigned to the assigned slot.", this));
        }

        try {

            // race double check, did we fail/cancel and do we need to release the slot?
            if (this.state != ExecutionState.DEPLOYING) {
                slot.releaseSlot(
                        new ERException(
                                "Actual state of execution "
                                        + this
                                        + " ("
                                        + state
                                        + ") does not match expected state DEPLOYING."));
                return;
            }

            LOG.info(
                    "Deploying {} to {} with allocation id {}",
                    executionId,
                    getAssignedResourceLocation(),
                    slot.getAllocationId());

            final CommandDescriptor cmd =
            		new CommandDescriptor(
            				        consumerId,
            				        command,
            						serArgs,
            						  executionId,
            						  slot.getAllocationId());

            final ProviderManagerGateway providerManagerGateway = slot.getProviderManagerGateway();

            // We run the submission in the future executor so that the serialization of large CDs
            // does not block
            // the main thread and sync back to the main thread once submission is completed.
            CompletableFuture.supplyAsync(
                            () -> providerManagerGateway.submitCommand(cmd, consumingMasterId, rpcTimeout), executor)
                    .thenCompose(Function.identity())
                    .whenCompleteAsync(
                            (ack, failure) -> {
                                if (failure == null) {
 
                                } else {
                                    final Throwable actualFailure =
                                            ExceptionUtils.stripCompletionException(failure);
                                    // 超时失败
                                    if (actualFailure instanceof TimeoutException) {
                                        markFailed(
                                                new Exception(
                                                        "Cannot deploy command "
                                                                + cmd.getCommand()
                                                                + " - TaskManager ("
                                                                + getAssignedResourceLocation()
                                                                + ") not responding after a rpcTimeout of "
                                                                + rpcTimeout,
                                                        actualFailure));
                                    } else {
                                        markFailed(actualFailure);
                                    }
                                }
                            },
                            mainThreadExecutor);

        } catch (Throwable t) {
            markFailed(t);
        }
    }

    public void cancel() {
        // depending on the previous state, we go directly to cancelled (no cancel call necessary)
        // -- or to canceling (cancel call needs to be sent to the task manager)

        // because of several possibly previous states, we need to again loop until we make a
        // successful atomic state transition
        assertRunningInResourceConsumerMainThread();
        while (true) {

            ExecutionState current = this.state;

            if (current == ExecutionState.CANCELING || current == ExecutionState.CANCELED) {
                // already taken care of, no need to cancel again
                return;
            }

            // these two are the common cases where we need to send a cancel call
            else if (current == ExecutionState.INITIALIZING || current == ExecutionState.RUNNING || current == ExecutionState.DEPLOYING) {
                // try to transition to canceling, if successful, send the cancel call
                if (startCancelling(NUM_CANCEL_CALL_TRIES)) {
                    return;
                }
                // else: fall through the loop
            } else if (current == ExecutionState.FINISHED) {
                // finished before it could be cancelled.
                // in any case, the task is removed from the TaskManager already

                // a pipelined partition whose consumer has never been deployed could still be
                // buffered on the TM
                // release it here since pipelined partitions for FINISHED executions aren't handled
                // elsewhere
                // covers the following cases:
                // 		a) restarts of this vertex
                // 		b) a global failure (which may result in a FAILED job state)
                return;
            } else if (current == ExecutionState.FAILED) {
                // failed before it could be cancelled.
                // in any case, the task is removed from the TaskManager already
                return;
            } else if (current == ExecutionState.CREATED || current == ExecutionState.SCHEDULED) {
                // from here, we can directly switch to cancelled, because no task has been deployed
                if (cancelAtomically()) {
                    return;
                }
                // else: fall through the loop
            } else {
                throw new IllegalStateException(current.name());
            }
        }
    }

    public CompletableFuture<?> suspend() {
        switch (state) {
            case RUNNING:
            case INITIALIZING:
            case DEPLOYING:
            case CREATED:
            case SCHEDULED:
                if (!cancelAtomically()) {
                    throw new IllegalStateException(
                            String.format(
                                    "Could not directly go to %s from %s.",
                                    ExecutionState.CANCELED.name(), state.name()));
                }
                break;
            case CANCELING:
                completeCancelling();
                break;
            case FINISHED:
                // a pipelined partition whose consumer has never been deployed could still be
                // buffered on the TM
                // release it here since pipelined partitions for FINISHED executions aren't handled
                // elsewhere
                // most notably, the TaskExecutor does not release pipelined partitions when
                // disconnecting from the JM
                break;
            case FAILED:
            case CANCELED:
                break;
            default:
                throw new IllegalStateException(state.name());
        }

        return releaseFuture;
    }

    /**
     * This method fails the vertex due to an external condition. The task will move to state
     * FAILED. If the task was in state RUNNING or DEPLOYING before, it will send a cancel call to
     * the TaskManager.
     *
     * @param t The exception that caused the task to fail.
     */
    public void fail(Throwable t) {
        processFail(t, true);
    }

    // --------------------------------------------------------------------------------------------
    //   Callbacks
    // --------------------------------------------------------------------------------------------

    /**
     * This method marks the task as failed, but will make no attempt to remove task execution from
     * the task manager. It is intended for cases where the task is known not to be running, or then
     * the TaskManager reports failure (in which case it has already removed the task).
     *
     * @param t The exception that caused the task to fail.
     */
    public void markFailed(Throwable t) {
        processFail(t, false);
    }

    void markFailed(
            Throwable t,
            boolean cancelTask) {
        processFail(t, cancelTask);
    }

    void markFinished() {

        assertRunningInResourceConsumerMainThread();

        // this call usually comes during RUNNING, but may also come while still in deploying (very
        // fast tasks!)
        while (true) {
            ExecutionState current = this.state;

            if (current == ExecutionState.INITIALIZING || current == ExecutionState.RUNNING || current == ExecutionState.DEPLOYING) {

                if (transitionState(current, ExecutionState.FINISHED)) {
                	releaseAssignedResource(null);
                    return;
                }
            } else if (current == ExecutionState.CANCELING) {
                // we sent a cancel call, and the task manager finished before it arrived. We
                // will never get a CANCELED call back from the job manager
                completeCancelling();
                return;
            } else if (current == ExecutionState.CANCELED || current == ExecutionState.FAILED) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Task FINISHED, but concurrently went to state " + state);
                }
                return;
            } else {
                // this should not happen, we need to fail this
                markFailed(
                        new Exception(
                                "Vertex received FINISHED message while being in state " + state));
                return;
            }
        }
    }

    private boolean cancelAtomically() {
        if (startCancelling(0)) {
            completeCancelling();
            return true;
        } else {
            return false;
        }
    }

    private boolean startCancelling(int numberCancelRetries) {
        if (transitionState(state, ExecutionState.CANCELING)) {
            providerManagerLocationFuture.cancel(false);
            sendCancelRpcCall(numberCancelRetries);
            return true;
        } else {
            return false;
        }
    }
    
    void completeCancelling() {

        // the taskmanagers can themselves cancel tasks without an external trigger, if they find
        // that the
        // network stack is canceled (for example by a failing / canceling receiver or sender
        // this is an artifact of the old network runtime, but for now we need to support task
        // transitions
        // from running directly to canceled

        while (true) {
            ExecutionState current = this.state;

            if (current == ExecutionState.CANCELED) {
                return;
            } else if (current == ExecutionState.CANCELING
                    || current == ExecutionState.RUNNING
                    || current == ExecutionState.INITIALIZING
                    || current == ExecutionState.DEPLOYING) {
            	
                if (transitionState(current, ExecutionState.CANCELED)) {
                    finishCancellation();
                    return;
                }

                // else fall through the loop
            } else {
                // failing in the meantime may happen and is no problem.
                // anything else is a serious problem !!!
                if (current != ExecutionState.FAILED) {
                    String message =
                            String.format(
                                    "Asynchronous race: Found %s in state %s after successful cancel call.",
                                    commandKey, state);
                    LOG.error(message);
                    
                }
                return;
            }
        }
    }

    private void finishCancellation() {
        releaseAssignedResource(new ResourceException("BusinessExecution " + this + " was cancelled."));
    }

    // --------------------------------------------------------------------------------------------
    //  Internal Actions
    // --------------------------------------------------------------------------------------------
    /**
     * Process a execution failure. The failure can be fired by JobManager or reported by
     * TaskManager. If it is fired by JobManager and the execution is already deployed, it needs to
     * send a PRC call to remove the task from TaskManager. It also needs to release the produced
     * partitions if it fails before deployed (because the partitions are possibly already created
     * in external shuffle service) or JobManager proactively fails it (in case that it finishes in
     * TaskManager when JobManager tries to fail it). The failure will be notified to SchedulerNG if
     * it is from within the ExecutionGraph. This is to trigger the failure handling of SchedulerNG
     * to recover this failed execution.
     *
     * @param t Failure cause
     * @param cancelTask Indicating whether to send a PRC call to remove task from TaskManager. True
     *     if the failure is fired by JobManager and the execution is already deployed. Otherwise it
     *     should be false.
     * @param userAccumulators User accumulators
     * @param metrics IO metrics
     * @param releasePartitions Indicating whether to release result partitions produced by this
     *     execution. False if the task is FAILED in TaskManager, otherwise true.
     * @param fromSchedulerNg Indicating whether the failure is from the SchedulerNg. It should be
     *     false if it is from within the ExecutionGraph.
     */
    private void processFail(
            Throwable t,
            boolean cancelTask) {

        assertRunningInResourceConsumerMainThread();

        ExecutionState current = this.state;

        if (current == ExecutionState.FAILED) {
            // already failed. It is enough to remember once that we failed (its sad enough)
            return;
        }

        if (current == ExecutionState.CANCELING) {
            completeCancelling();
            return;
        }

        checkState(transitionState(current, ExecutionState.FAILED, t));

        // success (in a manner of speaking)
        this.failureCause =
                Optional.of(
                        ErrorInfo.createErrorInfoWithNullableCause(t, getStateTimestamp(ExecutionState.FAILED)));
        releaseAssignedResource(t);

    }

    boolean switchToRecovering() {
        if (switchTo(ExecutionState.DEPLOYING, ExecutionState.INITIALIZING)) {
            return true;
        }

        return false;
    }

    boolean switchToRunning() {
        return switchTo(ExecutionState.INITIALIZING, ExecutionState.RUNNING);
    }

    private boolean switchTo(ExecutionState from, ExecutionState to) {

        if (transitionState(from, to)) {
            return true;
        } else {
            // something happened while the call was in progress.
            // it can mean:
            //  - canceling, while deployment was in progress. state is now canceling, or canceled,
            // if the response overtook
            //  - finishing (execution and finished call overtook the deployment answer, which is
            // possible and happens for fast tasks)
            //  - failed (execution, failure, and failure message overtook the deployment answer)

            ExecutionState currentState = this.state;

            if (currentState == ExecutionState.FINISHED || currentState == ExecutionState.CANCELED) {
                // do nothing, the task was really fast (nice)
                // or it was canceled really fast
            } else if (currentState == ExecutionState.CANCELING || currentState == ExecutionState.FAILED) {

            } else {
                String message =
                        String.format(
                                "Concurrent unexpected state transition of task %s from %s (expected %s) to %s while deployment was in progress.",
                                getExecutionId(), currentState, from, to);

                LOG.debug(message);

                // undo the deployment
                sendCancelRpcCall(NUM_CANCEL_CALL_TRIES);

                // record the failure
                markFailed(new Exception(message));
            }

            return false;
        }
    }

    /**
     * This method sends a CancelTask message to the instance of the assigned slot.
     *
     * <p>The sending is tried up to NUM_CANCEL_CALL_TRIES times.
     */
    private void sendCancelRpcCall(int numberRetries) {
        final LogicalSlot slot = assignedResource;

        if (slot != null) {
            final ProviderManagerGateway taskManagerGateway = slot.getProviderManagerGateway();

            CompletableFuture<Acknowledge> cancelResultFuture =
                    FutureUtils.retry(
                            () -> taskManagerGateway.cancelCommand(executionId, rpcTimeout),
                            numberRetries,
                            mainThreadExecutor);

            cancelResultFuture.whenComplete(
                    (ack, failure) -> {
                        if (failure != null) {
                            fail(new Exception("Task could not be canceled.", failure));
                        }
                    });
        }
    }

    /**
     * 释放获分配资源，获分配资源成功释放后，完成的释放futrue
     *
     * @param cause for the resource release, null if none
     */
    private void releaseAssignedResource(@Nullable Throwable cause) {

        assertRunningInResourceConsumerMainThread();

        final LogicalSlot slot = assignedResource;

        if (slot != null) {
            //
            slot.releaseSlot(cause)
                    .whenComplete(
                            (Object ignored, Throwable throwable) -> {
                                this.assertRunningInResourceConsumerMainThread();
                                if (throwable != null) {
                                    releaseFuture.completeExceptionally(throwable);
                                } else {
                                    releaseFuture.complete(null);
                                }
                            });
        } else {
            // no assigned resource --> we can directly complete the release future
            releaseFuture.complete(null);
        }
    }

    // --------------------------------------------------------------------------------------------
    //  Miscellaneous
    // --------------------------------------------------------------------------------------------

    public void transitionState(ExecutionState targetState) {
        transitionState(state, targetState);
    }

    private boolean transitionState(ExecutionState currentState, ExecutionState targetState) {
        return transitionState(currentState, targetState, null);
    }

    // 转换执行状态
    private boolean transitionState(
            ExecutionState currentState, ExecutionState targetState, Throwable error) {
        // sanity check
        if (currentState.isTerminal()) {
            throw new IllegalStateException(
                    "Cannot leave terminal state "
                            + currentState
                            + " to transition to "
                            + targetState
                            + '.');
        }

        if (state == currentState) {
            state = targetState;
            markTimestamp(currentState, targetState);

            if (error == null) {
                LOG.info(
                        "{} ({}) switched from {} to {}.",
                        this.getExecutionId(),
                        currentState,
                        targetState);
            } else if (LOG.isInfoEnabled()) {
                LOG.info(
                        "{} switched from {} to {} on {}.",
                        this.getExecutionId(),
                        currentState,
                        targetState,
                        getLocationInformation(),
                        ExceptionUtils.stripCompletionException(error));
            }

            if (targetState == ExecutionState.INITIALIZING || targetState == ExecutionState.RUNNING) {
                initializingOrRunningFuture.complete(null);
            } else if (targetState.isTerminal()) {
                // complete the terminal state future
                terminalStateFuture.complete(targetState);
            }

            return true;
        } else {
            return false;
        }
    }

    private String getLocationInformation() {
        if (assignedResource != null) {
            return assignedResource.getTaskManagerLocation().toString();
        } else {
            return "[unassigned resource]";
        }
    }

    private void markTimestamp(ExecutionState currentState, ExecutionState targetState) {
        long now = System.currentTimeMillis();
        markTimestamp(targetState, now);
        markEndTimestamp(currentState, now);
    }

    private void markTimestamp(ExecutionState state, long timestamp) {
        this.stateTimestamps[state.ordinal()] = timestamp;
    }

    private void markEndTimestamp(ExecutionState state, long timestamp) {
        this.stateEndTimestamps[state.ordinal()] = timestamp;
    }

    // ------------------------------------------------------------------------
    //  Standard utilities
    // ------------------------------------------------------------------------
    // 确认在主线程
    private void assertRunningInResourceConsumerMainThread() {
    	mainThreadExecutor.assertRunningInMainThread();
    }
}
